package com.warrior.superdata.service.impl;

import com.warrior.superdata.dds.DataSourceHelp;
import com.warrior.superdata.enums.DatasourceType;
import com.warrior.superdata.enums.OperateType;
import com.warrior.superdata.mysql.entity.dto.ApiInfoDetailDto;
import com.warrior.superdata.mysql.entity.dto.ApiParamDto;
import com.warrior.superdata.service.UpdateSqlService;
import com.warrior.superdata.util.CommonUtil;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.BulkByScrollTask;
import org.elasticsearch.index.reindex.UpdateByQueryRequest;
import org.elasticsearch.script.Script;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.List;

/**
 * @Description:
 * @ClassName: EsUpdateServiceImpl.java
 * @date: 2022/6/24 17:10
 * @author xujian
 */
@Service
@Slf4j
public class EsUpdateServiceImpl implements UpdateSqlService {

    @Autowired
    private DataSourceHelp dataSourceHelp;

    @Override
    public String getDatasourceType() {
        return DatasourceType.ES.getName();
    }

    @Override
    public String getOperateType() {
        return OperateType.UPDATE.getName();
    }

    @Override
    public Object execute(ApiInfoDetailDto apiInfoDetailDto) {
        Long result = 0l;
        try {
            UpdateByQueryRequest updateQuery = new UpdateByQueryRequest(apiInfoDetailDto.getMainTableName());
            List<ApiParamDto> params = apiInfoDetailDto.getParams();
            BoolQueryBuilder boolQueryBuilder = EsQueryHelpService.generateQuery(apiInfoDetailDto.getParams());
            StringBuffer sb = new StringBuffer();
            params.forEach(item -> {
                String fieldName = item.getParamTableFieldName().split("\\.")[1];
                if (1 == item.getIsUpdate()) {
                    if (CommonUtil.hasCastString(item.getParamJavaType())) {
                        sb.append("ctx._source.").append(fieldName).append("=").append("'").append(item.getParamValue()).append("'").append(";");
                    } else {
                        sb.append("ctx._source.").append(fieldName).append("=").append(item.getParamValue()).append(";");
                    }
                }
            });
            updateQuery.setQuery(boolQueryBuilder);
            updateQuery.setScript(new Script(sb.toString()));
            RestHighLevelClient restHighLevelClient = dataSourceHelp.getRestHighLevelClient(apiInfoDetailDto.getDatasourceDto());
            BulkByScrollResponse response = restHighLevelClient.updateByQuery(updateQuery, RequestOptions.DEFAULT);
            BulkByScrollTask.Status status = response.getStatus();
            result = status.getTotal();
        } catch (Exception e) {
            log.error("es更新异常 errMsg:{}", e.getMessage());
        }

        log.info("es update {} 条数据", result);
        return result;
    }


}
